package com.iwhale.sequence.redisidx.sequence;

import java.net.InetSocketAddress;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.iwhale.sequence.redisidx.config.RedisConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.redis.RedisArrayAggregator;
import io.netty.handler.codec.redis.RedisBulkStringAggregator;
import io.netty.handler.codec.redis.RedisDecoder;
import io.netty.handler.codec.redis.RedisEncoder;

@Component
public class SequenceServer {
    private static Logger logger = LoggerFactory.getLogger(SequenceServer.class);

    /**
     * netty内存泄漏检测 DISABLED 完全关闭内存泄露检测，并不建议 SIMPLE 以1%的抽样率检测是否泄露，默认级别 ADVANCED 抽样率同SIMPLE，但显示详细的泄露报告 PARANOID
     * 抽样率为100%，显示报告信息同ADVANCED
     */
    static {
        System.getProperties().put("io.netty.leakDetection.level", "DISABLED");
    }

    @Value("${sequence-server.server.host}")
    private String host;

    @Value("${sequence-server.server.port}")
    private int port;

    @Value("${sequence-server.server.useEpoll}")
    private boolean useEpoll;

    @Value("${sequence-server.sequence.delta}")
    private int delta;

    @Autowired
    private RedisConfig redisConfig;

    public SequenceServer() {

    }

    private EventLoopGroup makeBossGroup() {
        if (useEpoll) {
            return new EpollEventLoopGroup();
        }
        else {
            return new NioEventLoopGroup();
        }
    }

    private EventLoopGroup makeWorkerGroup() {
        if (useEpoll) {
            return new EpollEventLoopGroup();
        }
        else {
            return new NioEventLoopGroup();
        }
    }

    private Class<? extends ServerSocketChannel> getChannelClass() {
        if (useEpoll) {
            return EpollServerSocketChannel.class;
        }
        else {
            return NioServerSocketChannel.class;
        }
    }

    public void start() {
        EventLoopGroup bossGroup = makeBossGroup();
        EventLoopGroup workerGroup = makeWorkerGroup();
        InHandler inHandler = new InHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(getChannelClass()).localAddress(new InetSocketAddress(host, port))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        ChannelPipeline p = socketChannel.pipeline();
                        p.addLast(new RedisDecoder());
                        p.addLast(new RedisBulkStringAggregator());
                        p.addLast(new RedisArrayAggregator());
                        p.addLast(new RedisEncoder());
                        p.addLast(new InHandler());
                    }
                });
            ChannelFuture f = b.bind().addListener(future -> {
                if (future.isSuccess()) {
                    logger.info("******************************************************************");
                    logger.info("使用Epoll：{}", useEpoll);
                    logger.info("序列获取步长：{}", delta);
                    redisConfig.getHosts().forEach((name, hostInfo) -> {
                        logger.info("redis节点：{}:{}", name, hostInfo.getHost() + ":" + hostInfo.getPort());
                    });
                    logger.info("序列服务（{}:{}）启动成功！", host, port);
                    logger.info("******************************************************************");
                }
            }).sync();
            f.channel().closeFuture().sync();
        }
        catch (Exception e) {
            logger.error("序列服务启动失败！", e);
        }
        finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
